package com.example.hosrxjavaapplication.observable;

import com.example.hosrxjavaapplication.Constant;
import ohos.aafwk.ability.AbilitySlice;
import ohos.aafwk.ability.Lifecycle;
import ohos.app.AbilityContext;
import ohos.app.dispatcher.task.TaskPriority;
import ohos.hiviewdfx.HiLog;

import java.util.ArrayList;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/**
 * 通知者
 * create by chenjiayou on 2021/2/19
 */
public class ObservableImpl<T> implements Observable {

    AbilityContext context;
    Lifecycle lifecycle;
    String tag;
    Observer observer;//接收者对象
    ObservableOnSubscribe subscribe; //包含事件发射器的回调接口
    ObservableEmitter emitter; //事件发射器
    Schedulers observeOnSchedulers; //回调线程 UI线程
    Schedulers subscribeOnSchedulers; //执行线程 子线程
    long interval = 0L; //事件发送器间隔时间
    ArrayList<Work> works = new ArrayList();
    private Lock lock = new ReentrantLock();

    public ObservableImpl(AbilitySlice context, String tag, ObservableOnSubscribe<T> subscribe) {
        this.context = context;
        this.lifecycle = context.getLifecycle();
        this.tag = tag;
        this.subscribe = subscribe;
    }

    public ObservableImpl observeOn(Schedulers schedulers) {
        this.observeOnSchedulers = schedulers;
        return this;
    }

    public ObservableImpl subscribeOn(Schedulers schedulers) {
        this.subscribeOnSchedulers = schedulers;
        return this;
    }

    public ObservableImpl work(Work work) {
        works.add(work);
        return this;
    }

    public ObservableImpl interval(int interval) {
        this.interval = interval;
        return this;
    }

    public void subscribe(Observer observer) {
        this.observer = observer;
        this.emitter = new ObservableEmitter<T>() {

            @Override
            public void onNext(T t) {
                //执行线程
                doNext(t);
            }

            @Override
            public void onComplete() {
                //执行线程
                doComplete();
                //执行工作线程
                doWork();
            }
        };
        //装载事件发射器
        run(subscribeOnSchedulers, () -> this.subscribe.subscribe(emitter));
    }

    private void doWork() {
        //循环遍历工作线程 按照同步线程执行
        if (works != null && works.size() > 0) {
            final Object[] result = {null};
            works.forEach(work -> {
                result[0] = work.doWork(null == result[0] ? 0 : result[0]);
            });
            try {
                Thread.sleep(interval);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //切换到接收线程
            ObservableImpl.this.run(observeOnSchedulers, () -> {
                if (isActive()) {
                    observer.onSubscribe(result[0]);
                }
            });
        }
    }


    private void doNext(T t) {
        HiLog.info(Constant.label, "doNext => " + t + "  thread on " + Thread.currentThread());
        //接收线程
        if (isActive()) {
            try {
                Thread.sleep(interval);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //切换到接收线程
            ObservableImpl.this.run(observeOnSchedulers, () -> observer.onNext(t));
        }
    }

    private void doComplete() {
        HiLog.info(Constant.label, "doComplete => thread on " + Thread.currentThread());
        //接收线程
        if (isActive()) {
            try {
                Thread.sleep(interval);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            //切换到接收线程
            ObservableImpl.this.run(observeOnSchedulers, () -> observer.onComplete());
        }
    }

    private boolean isActive() {
        Lifecycle.Event state = lifecycle.getLifecycleState();
        return state != Lifecycle.Event.ON_STOP && state != Lifecycle.Event.ON_INACTIVE && state != Lifecycle.Event.ON_BACKGROUND && state != Lifecycle.Event.ON_FOREGROUND;
    }


    private void run(Schedulers schedulers, Runnable runnable) {
        if (schedulers == Schedulers.IO) {
            context.getGlobalTaskDispatcher(TaskPriority.DEFAULT).asyncDispatch(runnable);
        }
        if (schedulers == Schedulers.UI) {
            context.getUITaskDispatcher().syncDispatch(runnable);
        }
    }

}
